package org.pentaho.streaming.autotask;

import com.google.common.collect.ImmutableList;
import org.pentaho.di.trans.streaming.common.BlockingQueueStreamSource;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import static java.lang.Thread.sleep;
import static java.util.Collections.singletonList;

public class AutoTaskStreamSource extends BlockingQueueStreamSource<List<Object>> {
    private static final Class<?> PKG = AutoTaskStreamSource.class;
    private final AutoTaskConsumerMeta autoTaskConsumerMeta;
    private final AutoTaskConsumer autoTaskConsumer;

    private Thread thread;

    AutoTaskStreamSource(AutoTaskConsumerMeta AutoTaskConsumerMeta, AutoTaskConsumer AutoTaskConsumer) {
        super(AutoTaskConsumer);
        this.autoTaskConsumer = AutoTaskConsumer;
        this.autoTaskConsumerMeta = AutoTaskConsumerMeta;
    }

    @Override
    public void open() {
        thread = new Thread(() -> {
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS");
            while (true) {
                try {
                    acceptRows(singletonList(ImmutableList.of(simpleDateFormat.format(new Date()))));
                    sleep(Integer.parseInt(autoTaskConsumerMeta.getBatchDuration()));
                } catch (InterruptedException e) {
                    autoTaskConsumer.logError(e.toString());
                    break;
                }
            }
        });
        thread.start();
    }

    @Override
    public void close() {
        try {
            thread.interrupt();
            thread.join();
        } catch (InterruptedException e) {
            autoTaskConsumer.logError(e.toString());
        }
        super.close();
    }
}
